#include "udp_server.h"
#include "event_loop.h"
#include "channel.h"
using namespace el;


UdpServer::UdpServer(EventLoop * loop, SocketAddrIn & addr)
	:socket_(Socket::getUdpSocket()),
	is_start_(false),
	max_send_len_(1024)
{
	loop_ = loop;
	int rtn = socket_.bind(addr);
	channel_ = new Channel(loop_, socket_.fd());
	channel_->setReadCallBack(Functor(this, &UdpServer::handleRead));
	channel_->setWriteCallBack(Functor(this, &UdpServer::handleWrite));
	assert(rtn == 0);
}


UdpServer::~UdpServer()
{
	assert(loop_->isInLoopThread());
	assert(loop_->isProcessingPending());
	channel_->disableAllEvent();
	channel_->remove();
	delete channel_;
	socket_.closeFD();
}

void el::UdpServer::release()
{
	if (!loop_->isInLoopThread() || !loop_->isProcessingPending())
	{
		loop_->queueInLoop(Functor(this, &UdpServer::release));
	}
	else
	{
		delete this;
	}
}

bool el::UdpServer::isStart()
{
	assert(loop_->isInLoopThread());
	return is_start_;
}

void el::UdpServer::start()
{
	if (!loop_->isInLoopThread())
	{
		loop_->queueInLoop(Functor(this, &UdpServer::start));
	}
	else
	{
		channel_->enableReadEvent();
		is_start_ = true;
	}
}

void el::UdpServer::stop()
{
	if (!loop_->isInLoopThread())
	{
		loop_->queueInLoop(Functor(this, &UdpServer::stop));
	}
	else
	{
		channel_->disableAllEvent();
		is_start_ = false;
	}
}

void el::UdpServer::setMessageCallback(const UdpMessageCallback & callback)
{
	message_callback_ = callback;
}

void el::UdpServer::setErrorCallback(const UdpErrorCallback & callback)
{
	error_callback_ = callback;
}

void el::UdpServer::sendData(Buffer &buffer, const SocketAddrIn& addr)
{
	Buffer *send_buffer = new Buffer();
	buffer.swap(*send_buffer);
	write_queue_.push(UdpDataPair(addr, send_buffer));
	loop_->runInLoop(Functor(this, &UdpServer::sendInLoop));
}

int el::UdpServer::readData(Buffer& buffer, SocketAddrIn& addr)
{
	if (read_queue_.size() <= 0)
	{
		return 0;
	}
	UdpDataPair& read_data = read_queue_.takeFront();
	addr = read_data.first;
	int rtn = read_data.second->readableSize();
	if (buffer.empty())
	{
		buffer.swap(*read_data.second);
	}
	else
	{
		buffer.write(*read_data.second);
	}
	delete read_data.second;
	return rtn;
}

void el::UdpServer::handleRead()
{
	Buffer *buffer = new Buffer();
	SocketAddrIn addr;
	int read_count = ::recvfrom(socket_.fd(), buffer->peekWrite(), buffer->writableSize(), 0, 
		(sockaddr*)&addr.sockaddr(), &addr.lenght());
	if (read_count <= 0)
	{
		error_callback_(this, read_count);
		delete buffer;
	}
	else
	{
		buffer->write(read_count);
		read_queue_.push(UdpDataPair(addr, buffer));
		message_callback_(this);
	}
}

void el::UdpServer::handleWrite()
{
	sendTo(max_send_len_);
	if (write_queue_.size() == 0)
	{
		channel_->disableWriteEvent();
	}
}

void el::UdpServer::sendInLoop()
{
	if (channel_->hasWriteEvent())
	{
		return;
	}
	sendTo(max_send_len_);
	
	if (write_queue_.size() > 0)
	{
		channel_->enableWriteEvent();
	}
}

void el::UdpServer::sendTo(int max_len)
{
	if (write_queue_.size() == 0)
	{
		return;
	}
	SocketAddrIn addr = write_queue_.front().first;
	Buffer* buffer = write_queue_.front().second;

	int send_count = max_len > buffer->readableSize() ? buffer->readableSize() : max_len;
	int n = ::sendto(socket_.fd(), buffer->peekRead(), send_count, 0, (sockaddr*)&addr.sockaddr(), addr.lenght());
	if (n < 0)
	{
		int err = WSAGetLastError();
		error_callback_(this, err);
	}
	buffer->read(n);
	if (buffer->empty())
	{
		write_queue_.pop();
		if (write_queue_.size() == 0)
		{
			return;
		}
	}
	if (send_count == max_len)
	{
		return;
	}
	sendTo(max_len);
}
